package com.atguigu.userprofile.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.userprofile.bean.TagCondition;
import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.bean.UserGroup;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.mapper.UserGroupMapper;
import com.atguigu.userprofile.service.TagInfoService;
import com.atguigu.userprofile.service.UserGroupService;
import com.atguigu.userprofile.utils.RedisUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2021-05-04
 */
@Service
@Slf4j
@DS("mysql")
public class UserGroupServiceImpl extends ServiceImpl<UserGroupMapper, UserGroup> implements UserGroupService {

    @Autowired
    TagInfoService tagInfoService;

    @Override
    public void genUserGroupUids(UserGroup userGroup) {
        System.out.println(JSON.toJSONString(userGroup) );
        String bitmapAndSql = getBitmapAndSql(userGroup.getTagConditions(), userGroup.getBusiDate());
        System.out.println(bitmapAndSql);
        String insertSQL = insertSQL(bitmapAndSql, userGroup.getId().toString());
        baseMapper.insertSqlClickhouse(insertSQL);
        //查询人群包的人数
        String countSQL = countSQL(userGroup.getId().toString());
        Long userGroupCount = baseMapper.countSqlClickhouse(countSQL);
        refreshUserGroup(userGroupCount,userGroup);
        // 还要写入redis
        inserRedisSet(  userGroup.getId().toString());
        inserRedisBitmap(userGroup.getId().toString());
    }

    public void inserRedisSet(String userGroupId){
        // userGroup 人群包 uid的bitmap 集合
        //  1 type ?    set   2 key ?  user_group:[user_group_id] 如 user_group:101  3 value ? uids [123,4512,52,31]
        //    4 写入api ？sadd       ///  5 读取api ？ smembers

        // 1 查询出人群包 从clickhouse 中查询  list
        List<String> userIdList = baseMapper.selectUidList(userGroupId);
        // 2  把人群包写入redis
        Jedis jedis = new Jedis("bigdata01", 6379);
        String key="user_group:"+userGroupId;

        jedis.sadd(key, userIdList.toArray(new String[userIdList.size()]));

        jedis.close();

    }

    public void inserRedisBitmap(String userGroupId){
        // userGroup 人群包 uid的bitmap 集合
        //  1 type ?    set   2 key ?  user_group:[user_group_id] 如 user_group:101  3 value ? uids [123,4512,52,31]
        //    4 写入api ？sadd       ///  5 读取api ？ smembers

        // 1 查询出人群包 从clickhouse 中查询  list
        List<String> userIdList = baseMapper.selectUidList(userGroupId);
        // 2  把人群包写入redis
        Jedis jedis = RedisUtil.getJedisFromPool();
        String key="user_group_bit:"+userGroupId;
        //当本身某个api不支持批量提交的时候 ，可以使用pipeline  把多条命令组成一个批次一起提交
        Pipeline pipelined = jedis.pipelined();
        for (String uid  : userIdList) {
            pipelined.setbit(key,Long.valueOf(uid),true );
        }
        pipelined.sync();
        jedis.close(); // 如果连接时new创建close直接释放  如果是从连接池中借来的，close会还到连接池。

    }


    //  人群包
    @Override
    public Long countUserGroupUids(UserGroup userGroup) {

        String bitmapAndSql = getBitmapAndSql(userGroup.getTagConditions(), userGroup.getBusiDate());
        String coutSQL=" select  bitmapCardinality( "+bitmapAndSql+")";
        Long userGroupCount = baseMapper.countSqlClickhouse(coutSQL);
        return userGroupCount;
    }

    @Override
    public void refreshUserGroupUids(String userGroupId,String busiDate) {
        //1  根据id把人群的基本信息查询出来 mysql
        UserGroup userGroup = getById(userGroupId);
        //2  把旧的人群包的bitmap数据 删除 clickhouse
        baseMapper.deleteUserGroupById(userGroupId);
        //3  利用条件重新 执行人群包的写入bitmap
        //把数据库中的条件json转为List对象
        String conditionJsonStr = userGroup.getConditionJsonStr();
        List<TagCondition> tagConditionList = JSON.parseArray(conditionJsonStr, TagCondition.class);
        userGroup.setTagConditions(tagConditionList);
        //把busiDate补入对象
        userGroup.setBusiDate(busiDate);
        genUserGroupUids( userGroup);

    }

    //更新人群数 和 刷新更新时间
    private void refreshUserGroup(Long userGroupCount,UserGroup userGroup){
        userGroup.setUpdateTime(new Date());
        userGroup.setUserGroupNum(userGroupCount);
        saveOrUpdate(userGroup);
    }

    private String insertSQL(String bitmapAndSql ,String userGroupId){
        String insertSQL="insert into user_group select '"+userGroupId+"',("+ bitmapAndSql+")";
        return insertSQL;
    }

    //bitmapCardinality 返回bitmap保存值的个数
    private String countSQL( String userGroupId){
        String coutSQL=" select  bitmapCardinality(us )  from user_group where user_group_id='"+userGroupId+"'";
        return coutSQL;
    }

    /*
     条件   1 tag_code  2  tag_value   3  dt  4 表名  5 判断符号 6  多个值 要进行拼接 7 要判断值的类型  单引
	 需要   1 tagCode （有）
			2 tagValues 加工 （1是否多值 2 是否需要加单引) --> 标签值类型 -->查标签元数据(mysql tag_info)
			   数组解析
			3 dt -> busiDate (有）
			4 表名 --> 标签值类型 -->查标签元数据(mysql tag_info)  靠标签值类型判断获得
			5 判断符号  -> operator 匹配 eq : =    lte :<=
			6、7同3
     */
/*
   select groupBitmapMergeState(us)  from user_tag_value_long
  where tag_code='TG_BEHAVIOR_BUY_ORDER30CT'  and tag_value >= 1 and dt='2021-05-16'

 */
    private String genConditionSql(TagCondition tagCondition,String busiDate,String tagValueType){
        String tagCode=tagCondition.getTagCode();
        String tagValue=   getValueSql(tagCondition.getTagValues(),  tagValueType );
        String tableName = getTableName(tagValueType);
        String operatorStr = getOperatorStr(tagCondition.getOperator());

        String conditionSql =new StringBuilder("  select groupBitmapMergeState(us)  " )
                .append(" from  ").append(tableName)
                .append(" where tag_code='").append(tagCode).append("'")
                .append (" and tag_value ").append(operatorStr).append(" ").append(tagValue)
                .append("  and  dt='").append(busiDate).append("'").toString();

         return conditionSql ;

    }

    //获得标签值
    private String  getValueSql(List<String> tagValueList ,String tagValueType){
        String finalTagvalue="";
        int i =0;
        for (String tagValue : tagValueList) {
            if(i++ >0){
                finalTagvalue=finalTagvalue+",";
            }
            if(tagValueType.equals(ConstCodes.TAG_VALUE_TYPE_STRING)||tagValueType.equals( ConstCodes.TAG_VALUE_TYPE_DATE)) {
                finalTagvalue += "'" + tagValue + "'";
            }else{
                finalTagvalue +=    tagValue  ;
            }

        }
        if(tagValueList.size()>1){
            finalTagvalue= "("+finalTagvalue+")";
        }
        return finalTagvalue;
    }

    private  String  getTableName(String tagValueType){
         String tableName=null;

        if(tagValueType.equals(ConstCodes.TAG_VALUE_TYPE_STRING)  ){
            tableName ="user_tag_value_string";
        }else if( tagValueType.equals(ConstCodes.TAG_VALUE_TYPE_DATE)  ){
            tableName ="user_tag_value_date";
        }else if( tagValueType.equals( ConstCodes.TAG_VALUE_TYPE_LONG) ){
            tableName ="user_tag_value_long";
        }else if( tagValueType.equals( ConstCodes.TAG_VALUE_TYPE_DECIMAL )){
            tableName ="user_tag_value_decimal";
        }
        return tableName;
  }
    private String getOperatorStr(String operatorShortName){
        switch (operatorShortName){
            case "eq":
                return "=";
            case "lte":
                return "<=";
            case "gte":
                return ">=";
            case "lt":
                return "<";
            case "gt":
                return ">";
            case "neq":
                return "<>";
            case "in":
                return "in";
            case "nin":
                return "not in";
        }
        throw  new RuntimeException("操作符不正确");

    }


    //   bitand ( bitand ( sql1 ,sql2) ,sql3)
    public String  getBitmapAndSql(List<TagCondition> tagConditionList ,String busiDate){
        Map<String, TagInfo> tagInfoMapWithCode = tagInfoService.getTagInfoMapWithCode();
        String bitmapAndSql="";
        for (TagCondition tagCondition : tagConditionList) {
            String tagCode = tagCondition.getTagCode();
            TagInfo tagInfo = tagInfoMapWithCode.get(tagCode);
            String conditionSql = genConditionSql(tagCondition, busiDate, tagInfo.getTagValueType());
            if(bitmapAndSql.length() ==0){
                bitmapAndSql="("+conditionSql+")";
            }else{
                bitmapAndSql="bitmapAnd ("+bitmapAndSql+",("+conditionSql+"))";
            }

        }
        return bitmapAndSql;
    }
}
